//
// Created by syj on 2023/10/10.
//

#ifndef HELLO_DDS_DDS_RECEIVER_H
#define HELLO_DDS_DDS_RECEIVER_H

#include <iostream>
#include <thread>
#include <memory>
#include <atomic>
#include <condition_variable>
#include <mutex>
#include <map>

#include <fastdds/dds/domain/DomainParticipant.hpp>
#include <fastdds/dds/subscriber/DataReaderListener.hpp>
//#include <fastrtps/subscriber/SampleInfo.h>
#include <fastdds/dds/core/status/SubscriptionMatchedStatus.hpp>
#include <fastdds/dds/subscriber/SampleInfo.hpp>

#include <csignal>
#include <fastdds/rtps/transport/shared_mem/SharedMemTransportDescriptor.h>
#include <fastdds/rtps/transport/UDPv4TransportDescriptor.h>
#include <fastdds/rtps/transport/UDPv6TransportDescriptor.h>

#include <fastrtps/attributes/ParticipantAttributes.h>
#include <fastrtps/attributes/SubscriberAttributes.h>
#include <fastdds/dds/domain/DomainParticipantFactory.hpp>
#include <fastdds/dds/subscriber/Subscriber.hpp>
#include <fastdds/dds/subscriber/DataReader.hpp>
#include <fastdds/dds/subscriber/SampleInfo.hpp>
#include <fastdds/dds/subscriber/qos/DataReaderQos.hpp>

#include "transport/receiver/receiver.h"

#include "transport/transmitter/transmitter.h"
#include "transport/dds/DDSMessagePubSubTypes.h"
#include "transport/dds/DDSMessagePubSub.hpp"
namespace motovis {
    namespace transport {
        template<typename MessageT>
        class DDSReceiver: public Receiver<MessageT>
        {
        public:
            using MessagePtr = std::shared_ptr<MessageT>;
            using MessageListener = std::function<void(const MessagePtr& msg)>;

            DDSReceiver()
            :type_(new MessagePubSubType<MessageT>())
            {
                DDSPublisher::Instance()->init("DataReader");
                init();
            }
            virtual ~DDSReceiver()
            {
                if (DDSPublisher::Instance()->GetParticipant() != nullptr)
                {
                    if (DDSPublisher::Instance()->GetSubscriber() != nullptr)
                    {
//            if (reader_ != nullptr)
                        {
//                DDSPublisher::Instance()->GetSubscriber()->delete_datareader(reader_);
                        }
                        for (auto& item: m_reader_)
                        {
                            DDSPublisher::Instance()->GetSubscriber()->delete_datareader(item.second);
                        }
//                        DDSPublisher::Instance()->GetParticipant()->delete_subscriber(DDSPublisher::Instance()->GetSubscriber());
                    }
//        if (topic_ != nullptr)
                    {
//            DDSPublisher::Instance()->GetParticipant()->delete_topic(topic_);
                    }
                    for (auto& item: m_topic_)
                    {
                        DDSPublisher::Instance()->GetParticipant()->delete_topic(item.second);
                    }
//                    eprosima::fastdds::dds::DomainParticipantFactory::get_instance()->delete_participant(DDSPublisher::Instance()->GetParticipant());
                }
            }
            /* channel_name: topic_name
             * msg_listener: 接收数据的回调指针, 入参为数据智能指针
             * */
            void addListener(const std::string& channel_name, const MessageListener& msg_listener) override
            {
                creatReader(channel_name,1);
                listener_.m_func = msg_listener;
            }

            //! Initialize the publisher
            /* domain: domainID
             * */
            bool init()

            {
                // REGISTER THE TYPE
                type_.register_type(DDSPublisher::Instance()->GetParticipant());

                return true;
            }

            bool creatReader(const std::string& topic_name,/*const std::string& message,*/uint32_t max_messages,
                             DDSTransportType transport=DDSTransportType::SHM,
                             bool reliable=false,
                             bool transient=false,
//                     const std::string& partitions = "",
                             bool use_ownership = false,
                             const std::string& profile = "")

            {
                if (nullptr == DDSPublisher::Instance()->GetParticipant() || nullptr == DDSPublisher::Instance()->GetSubscriber())
                {
                    return false;
                }

                eprosima::fastdds::dds::DataReaderQos rqos = eprosima::fastdds::dds::DATAREADER_QOS_DEFAULT;

                // Data sharing set in endpoint. If it is not default, set it to off
                if (transport != DDSTransportType::DEFAULT)
                {
                    rqos.data_sharing().off();
                }
                else
                {
                    rqos.data_sharing().automatic();  // default
                }

                if (reliable)
                {
                    rqos.reliability().kind = eprosima::fastdds::dds::RELIABLE_RELIABILITY_QOS;
                    rqos.history().kind = eprosima::fastdds::dds::KEEP_ALL_HISTORY_QOS;
                }
                else
                {
                    rqos.reliability().kind = eprosima::fastdds::dds::BEST_EFFORT_RELIABILITY_QOS;  // default
                }

                if (transient)
                {
                    rqos.durability().kind = eprosima::fastdds::dds::TRANSIENT_LOCAL_DURABILITY_QOS;
                }
                else
                {
                    rqos.durability().kind = eprosima::fastdds::dds::VOLATILE_DURABILITY_QOS;   // default
                }

                // Set ownership
                if (use_ownership)
                {
                    rqos.ownership().kind = eprosima::fastdds::dds::OwnershipQosPolicyKind::EXCLUSIVE_OWNERSHIP_QOS;
                }

                /* 创建具体的订阅端口 */
                eprosima::fastdds::dds::Topic* pTopic{nullptr};
                eprosima::fastdds::dds::DataReader* pReader{nullptr};

                // CREATE THE TOPIC
                // CREATE THE READER
                if (max_messages > 0)
                {
                    listener_.set_max_messages(max_messages);
                }
                auto topicItr = m_topic_.find(topic_name);
                if (topicItr != m_topic_.end())
                {
                    pTopic = topicItr->second;
                    if (pTopic == nullptr)
                    {
                        return false;
                    }
                    auto readerItr = m_reader_.find(pTopic);
                    if (readerItr != m_reader_.end())
                    {
                        pReader = readerItr->second;
                        if (pReader == nullptr)
                        {
                            return false;
                        }
                    }
                    else
                    {
                        /* 创建DataWriter */
                        pReader = DDSPublisher::Instance()->GetSubscriber()->create_datareader(pTopic, rqos, &listener_);
                        if (pReader == nullptr)
                        {
                            return false;
                        }
                        m_reader_.insert(std::pair<eprosima::fastdds::dds::Topic*,eprosima::fastdds::dds::DataReader*>{pTopic,pReader});
                    }
                }
                else
                {
                    // CREATE THE TOPIC
                    pTopic = DDSPublisher::Instance()->GetParticipant()->create_topic(topic_name, "Motovis", eprosima::fastdds::dds::TOPIC_QOS_DEFAULT);
                    if (pTopic == nullptr)
                    {
                        return false;
                    }
                    pReader = DDSPublisher::Instance()->GetSubscriber()->create_datareader(pTopic, rqos, &listener_);
                    if (pReader == nullptr)
                    {
                        return false;
                    }
                    m_topic_.insert(std::pair<std::string,eprosima::fastdds::dds::Topic*>{topic_name,pTopic});
                    m_reader_.insert(std::pair<eprosima::fastdds::dds::Topic*,eprosima::fastdds::dds::DataReader*>{pTopic,pReader});
                }

                std::cout << "Subscriber Participant created with DataReader Guid [ " /*<< reader_->guid() <<*/ " ]." << std::endl;


                stop_ = false;
                signal(SIGINT, [](int signum)
                {
                    std::cout << "SIGINT received, stopping Subscriber execution." << std::endl;
                    static_cast<void>(signum);
                    DDSReceiver::stop();
                });
                std::unique_lock<std::mutex> lck(terminate_cv_mtx_);
                terminate_cv_.wait(lck, []
                {
                    return is_stopped();
                });

                return true;
            }

            //! Return the current state of execution
            static bool is_stopped()
            {
                return stop_;
            }

            //! Trigger the end of execution
            static void stop()
            {
                stop_ = true;
                terminate_cv_.notify_all();
            }
        private:
            eprosima::fastdds::dds::TypeSupport type_;
            std::map<std::string,eprosima::fastdds::dds::Topic*> m_topic_;
            std::map<eprosima::fastdds::dds::Topic*,
                    eprosima::fastdds::dds::DataReader*> m_reader_;
            class SubListener : public eprosima::fastdds::dds::DataReaderListener
            {
            public:

                SubListener()
                        : matched_(0)
                        , samples_(0)
                        , max_messages_(0)
                        ,p_msg(new MessageT)
                {
                }

                ~SubListener() override
                {
                }

                //! Set the maximum number of messages to receive before exiting
                void set_max_messages(uint32_t max_messages)
                {
                    max_messages_ = max_messages;
                }
                //! Callback executed when a new sample is received
                void on_data_available(
                        eprosima::fastdds::dds::DataReader* reader) override
                {
                    eprosima::fastdds::dds::SampleInfo info;
                    while ((reader->take_next_sample(p_msg.get(), &info) == ReturnCode_t::RETCODE_OK) && !is_stopped())
                    {
                        if (info.instance_state == eprosima::fastdds::dds::ALIVE_INSTANCE_STATE)
                        {
                            samples_++;
                            // Print your structure data here.
                            if (nullptr!= m_func)
                            {
                                m_func(p_msg);
                            }
                            std::cout << "Message " << " RECEIVED" << std::endl;
                            if (max_messages_ > 0 && (samples_ >= max_messages_))
                            {
                                DDSReceiver::stop();
                            }
                        }
                    }
                }
                //! Callback executed when a DataWriter is matched or unmatched
                void on_subscription_matched(
                        eprosima::fastdds::dds::DataReader* reader,
                        const eprosima::fastdds::dds::SubscriptionMatchedStatus& info) override
                {
                    if (info.current_count_change == 1)
                    {
                        matched_ = info.current_count;
                        std::cout << info.current_count << " Subscriber matched [ " << iHandle2GUID(info.last_publication_handle) << " ]." << std::endl;
                    }
                    else if (info.current_count_change == -1)
                    {
                        matched_ = info.current_count;
                        std::cout << info.current_count << " Subscriber unmatched [ " << iHandle2GUID(info.last_publication_handle) << " ]." << std::endl;
                    }
                    else
                    {
                        std::cout << info.current_count_change
                                  << " is not a valid value for SubscriptionMatchedStatus current count change" << std::endl;
                    }
                }

                MessageListener m_func;
            private:
                MessagePtr p_msg;

                //! Number of DataWriters matched to the associated DataReader
                int matched_; /* topic 对应 writer 的个数 */

                //! Number of samples received
                uint32_t samples_;

                //! Number of messages to be received before triggering termination of execution
                uint32_t max_messages_;
            }
                    listener_;
            //! Member used for control flow purposes
            static std::atomic<bool> stop_;

            //! Protects terminate condition variable
            static std::mutex terminate_cv_mtx_;

            //! Waits during execution until SIGINT or max_messages_ samples are received
            static std::condition_variable terminate_cv_;
        };
        template<typename MessageT>
        std::atomic<bool> DDSReceiver<MessageT>::stop_(false);
        template<typename MessageT>
        std::mutex DDSReceiver<MessageT>::terminate_cv_mtx_;
        template<typename MessageT>
        std::condition_variable DDSReceiver<MessageT>::terminate_cv_;

    }
}

#endif //HELLO_DDS_DDS_RECEIVER_H
